package com.raycloud.manager.msgqueue.impl;

import java.util.Date;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;

import javax.servlet.ServletContext;

import org.apache.log4j.Logger;
import org.springframework.web.context.ServletContextAware;

import com.raycloud.common.ScopeConstants;
import com.raycloud.dao.MsgIndexDao;
import com.raycloud.dao.UserRelationDao;
import com.raycloud.dao.dbhash.DbHashUtil;
import com.raycloud.domain.MsgDistributeJob;
import com.raycloud.domain.MsgIndex;
import com.raycloud.domain.UserRelation;
import com.raycloud.manager.msgqueue.IMsgQueueHandler;

/**
 * 消息队列的消费者，负责把消息插入消息队列，此处实现较为简单只是为了描述生产者-消费者模型中的消费者， 后面随着消息队列的功能健壮会越来越复杂
 * 
 * @author wangyi
 * 
 */
public class MsgQueueHandler implements IMsgQueueHandler, ServletContextAware {
	private static final Logger logger = Logger
			.getLogger(MsgQueueHandler.class);

	private ServletContext sc;

	private UserRelationDao userRelationDao;
	private MsgIndexDao msgIndexDao;

	@SuppressWarnings("unchecked")
	public void execute() {
		ConcurrentLinkedQueue<MsgDistributeJob> jobQueue = (ConcurrentLinkedQueue<MsgDistributeJob>) sc
				.getAttribute(ScopeConstants.JOB_QUEUE);
		MsgDistributeJob job = null;
		while (true) {
			try {
				job = jobQueue.poll();
				if (job != null) {
					distribute(job);
				} else {
					//logger.info("没有任务，暂停一会儿");
					Thread.sleep(10 * 1000);
				}
			} catch (Exception e) {
				logger.error(e.getMessage(), e);
			}
		}
	}

	public void distribute(MsgDistributeJob job) throws Exception {
		try {
			long userId = job.getUserId();
			List<UserRelation> allFans = userRelationDao.queryAllFans(
					DbHashUtil.getUserRelationDBId(userId), userId);
			if (allFans != null && allFans.size() > 0) {
				logger.info("用户 id = " + userId + " 一共有" + allFans.size() + "粉丝要推送");
				for (UserRelation fans : allFans) {
					try {
						MsgIndex fansMsgIndex = new MsgIndex(
								fans.getFollowId(), userId, job.getMsgId(),
								new Date());
						fansMsgIndex.setDbId(DbHashUtil.getMsgIndex(fans.getFollowId()));
						msgIndexDao.insertMsgIndex(fansMsgIndex);
					} catch (Exception e) {
						logger.error(
								"向用户分发消息失败 follow_id=" + fans.getFollowId()
										+ " user_id=" + fans.getUserId(), e);
					}
				}
			}
		} catch (Exception e) {
			logger.error(e.getMessage(), e);
			throw new Exception(e);
		}
	}

	public void setServletContext(ServletContext sc) {
		this.sc = sc;
	}

	public void setUserRelationDao(UserRelationDao userRelationDao) {
		this.userRelationDao = userRelationDao;
	}

	public void setMsgIndexDao(MsgIndexDao msgIndexDao) {
		this.msgIndexDao = msgIndexDao;
	}
}
